package com.nd.bdc.presto.util;

import com.nd.bdc.presto.connection.PrestoJdbcClient;
import com.nd.bdc.presto.entity.PrestoSelectResult;
import com.nd.bdc.presto.entity.QueryResultSet;
import com.nd.bdc.presto.entity.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

@Component
public class SqlSelectManager {
    private static final Logger LOGGER = LoggerFactory.getLogger(PrestoJdbcClient.class);


    public PrestoSelectResult select(Connection cn, String sql, int startRow, int fetchSize)
            throws SQLException {
        QueryResultSet qrs = null;

        try {
            qrs = executeQuery(cn, sql, ResultSet.TYPE_FORWARD_ONLY);
            ResultSet rs = qrs.getResultSet();

            List<Row> rows = mapToRows(rs.getMetaData(), rs, startRow, fetchSize);

            return new PrestoSelectResult(sql, rs.getMetaData(), startRow, fetchSize, rows);
        } finally {
            QueryResultSet.close(qrs);
        }
    }

    public PrestoSelectResult select(String sql, ResultSet rs, int startRow, int fetchSize) throws SQLException {

        List<Row> rows = mapToRows(rs.getMetaData(), rs, startRow, fetchSize);
        return new PrestoSelectResult(sql, rs.getMetaData(), startRow, fetchSize, rows);
    }

    public QueryResultSet executeQuery(Connection cn, String sql, int resultSetType) throws SQLException {
        LOGGER.debug("execute {}, resultSetType={}", sql, resultSetType);

        PreparedStatement pst = null;
        ResultSet rs = null;
        List<Object> setParams = null;

        try {
            pst = createQueryPreparedStatement(cn, sql, resultSetType);
            rs = pst.executeQuery();

            return new QueryResultSet(pst, rs, setParams);
        } catch (SQLSyntaxErrorException | SQLDataException | SQLTimeoutException | SQLWarning e) {
            // "这些异常必定不是驱动程序的ResultSet.TYPE_SCROLL_*支持与否问题，不需要再降级处理"
            SQLException e1 = e;
            throw e1;
        } catch (SQLException e) {
            //("某些不支持ResultSet.TYPE_SCROLL_*的驱动程序不是在创建PreparedStatemen时报错，"
            //        + "而是在执行SQL的时候，所以在这里检查，必要时降级重新执行查询")
            int actualResultSetType = pst.getResultSetType();

            if (ResultSet.TYPE_FORWARD_ONLY == actualResultSetType) {
                IOUtil.closeIf(setParams);
                JdbcUtil.closeResultSet(rs);
                JdbcUtil.closeStatement(pst);

                throw e;
            } else {
                JdbcUtil.closeResultSet(rs);
                JdbcUtil.closeStatement(pst);

                LOGGER.debug("query is downgraded to [ResultSet.TYPE_FORWARD_ONLY] for exception :", e);

                //("降级为ResultSet.TYPE_FORWARD_ONLY重新执行")
                QueryResultSet qrs = executeQuery(cn, sql, ResultSet.TYPE_FORWARD_ONLY);
                return qrs;
            }
        }
    }

    public PreparedStatement createQueryPreparedStatement(Connection cn, String sql, int resultSetType)
            throws SQLException {
        if (ResultSet.TYPE_FORWARD_ONLY == resultSetType) {
            return cn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
        } else {
            PreparedStatement pst = null;

            try {
                pst = cn.prepareStatement(sql, resultSetType, ResultSet.CONCUR_READ_ONLY);
            } catch (SQLFeatureNotSupportedException e) {
                LOGGER.debug(
                        "query PreparedStatement is downgraded to [ResultSet.TYPE_FORWARD_ONLY] for exception :" + e);

                //("某些驱动程序不支持TYPE_SCROLL_INSENSITIVE，则降级获取")
                PreparedStatement thenPst = cn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY,
                        ResultSet.CONCUR_READ_ONLY);
                pst = thenPst;
            }

            return pst;
        }
    }

    protected List<Row> mapToRows(ResultSetMetaData metaData, ResultSet rs, int startRow, int count)
            throws SQLException {
        if (startRow < 1) {
            startRow = 1;
        }

        List<Row> resultList = new ArrayList<>();

        if (count >= 0 && startRow > 1) {
            forwardBefore(rs, startRow);
        }

        int endRow = (count >= 0 ? startRow + count : -1);

        int rowIndex = startRow;
        while (rs.next()) {
            if (endRow >= 0 && rowIndex >= endRow) {
                break;
            }

            Row row = mapToRow(metaData, rs);

            resultList.add(row);

            rowIndex++;
        }

        return resultList;
    }

    public void forwardBefore(ResultSet rs, int rowIndex) throws SQLException {
        // 第一行不做任何操作，避免不必要的调用可能导致底层不支持而报错
        if (rowIndex == 1) {
            return;
        }

        try {
            rs.absolute(rowIndex - 1);
        } catch (SQLException e) {
            LOGGER.debug("ResultSet [absolute()] is downgraded to [next()] for exception :", e);

            //("避免驱动程序或者ResultSet不支持absolute而抛出异常")
            int i = 1;
            for (; i < rowIndex; i++) {
                if (!rs.next()) {
                    break;
                }
            }
        }
    }

    public Row mapToRow(ResultSetMetaData metaData, ResultSet rs)
            throws SQLException {
        Row row = new Row();
        int colsLen = metaData.getColumnCount();
        for (int i = 0; i < colsLen; i++) {
            String colsName = metaData.getColumnLabel(i + 1);
            Object colsValue = rs.getObject(colsName);

            row.put(colsName, colsValue);
        }

        return row;

    }

}
